package com.doit.shark.dsp.kpirpts

import java.util.Properties

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/9 10:52
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:
  **/
object AreaKpi {

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession.builder().appName(AreaKpi.getClass.getSimpleName).master("local[*]").getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 加载数据
    val df = spark.read.parquet("G:\\sharkdata\\dsppreout")

    // DSL 风格
    val res = df.groupBy("provincename","cityname")
        .agg(
          sum(when( 'requestmode === "1" and 'processnode >= "1",1).otherwise(0)).as("ys_req_cnt"),
          sum(when( 'requestmode === "1" and 'processnode >= "2",1).otherwise(0)).as("yx_req_cnt"),
          sum(when( 'requestmode === "1" and 'processnode === "3",1).otherwise(0)).as("ad_req_cnt"),
          sum(when( 'iseffective === "1" and 'isbilling === "1" and 'isbid === "1" and 'adorderid =!= "0",1).otherwise(0)).as("bid_req_cnt"),
          sum(when( 'iseffective === "1" and 'isbilling === "1" and 'iswin ==="1",1).otherwise(0)).as("bid_succ_cnt"),
          sum(when( 'requestmode === "2" and 'iseffective === "1",1).otherwise(0)).as("ad_show_cnt"),
          sum(when( 'requestmode === "3" and 'iseffective === "1",1).otherwise(0)).as("ad_click_cnt"),
          sum(when( 'iseffective === "1" and 'isbilling === "1" and 'iswin ==="1",'winprice/1000d).otherwise(0)).as("ad_consume_cnt"),
          sum(when( 'iseffective === "1" and 'isbilling === "1" and 'iswin ==="1",'adpayment/1000d).otherwise(0)).as("ad_cost_cnt")
    )


    // SQL风格
    df.createTempView("tmp")

    val res2 = spark.sql(
      """
        |
        |select
        |
        |'2019-04-08' as dt,
        |provincename,
        |cityname,
        |
        |sum(case when requestmode = '1' and processnode >= '1' then 1 else 0 end) as ys_req_cnt,
        |sum(case when requestmode = '1' and processnode >= '2' then 1 else 0 end) as yx_req_cnt,
        |sum(case when requestmode = '1' and processnode = '3' then 1 else 0 end) as ad_req_cnt,
        |sum(case when iseffective = '1' and isbilling = 1 and isbid = '1' and adorderid != 0 then 1 else 0 end) as bid_req_cnt,
        |sum(case when iseffective = '1' and isbilling = 1 and iswin = '1' then 1 else 0 end) as bid_succ_cnt,
        |sum(case when requestmode = '2' and iseffective = '1' then 1 else 0 end) as ad_show_cnt,
        |sum(case when requestmode = '3' and iseffective = '1' then 1 else 0 end) as ad_click_cnt,
        |sum(case when iseffective = '1' and isbilling = '1' and iswin ='1' then cast(winprice as double)/1000 else 0 end) as ad_consume_cnt,
        |sum(case when iseffective = '1' and isbilling = '1' and iswin ='1' then cast(adpayment as double)/1000 else 0 end) as ad_cost_cnt
        |
        |from tmp
        |group by provincename,cityname
        |
      """.stripMargin)

    res.show(10,false)
    res2.show(10,false)


    val props = new Properties()
    props.setProperty("user","root")
    props.setProperty("password","Root.123")

    res2.write.mode(SaveMode.Append).jdbc("jdbc:mysql://c701:3306/doit_mall?charactorEncoding=utf8","dsp_kpi_area",props)

    spark.close()
  }

}
